package com.dahua.messagesyncsh.kafka;

import com.alibaba.fastjson2.JSONObject;
import com.dahua.messagesyncsh.config.redis.RedisUtil;
import com.dahua.messagesyncsh.pojo.SyncData;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.EnableAsync;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

import java.util.Map;

import static com.dahua.messagesyncsh.common.Constants.CMODE_FEATURE_SYNC;

@Slf4j
@Component
@EnableAsync
public class Schedule {
    @Resource
    private RedisUtil redisUtil;

    @Scheduled(cron = "0 0/10 * * * ?")
    public void updateEquipmentInfo() {
        Map<Object, Object> hash = redisUtil.getHash(CMODE_FEATURE_SYNC);

        long boundary = System.currentTimeMillis() - (30 * 60 * 1000L);

        for (Object key : hash.keySet()) {
            try{
                String keyStr = key.toString();
                SyncData syncData = JSONObject.parseObject(JSONObject.toJSONString(hash.get(key)), SyncData.class);

                if (syncData == null) {
                    log.info("【KafkaSchedule】：空key为{}", keyStr);
                    redisUtil.deleteHash(CMODE_FEATURE_SYNC, keyStr);
                    continue;
                }

                if (syncData.getStartTime() < boundary) {
                    redisUtil.deleteHash(CMODE_FEATURE_SYNC, keyStr);
                    log.info("【KafkaSchedule】：已删key为{}", keyStr);
                }
            }catch (Exception e){
                log.error("【KafkaSchedule】：{}", e.getMessage());
            }

        }
        log.info("【KafkaSchedule】：CMODE_FEATURE_SYNC定时任务更新");
    }
}
